#ifndef _PAXOS_H
#define _PAXOS_H

#include "messaging.h"
#include "paxos_protocol.pb.h"
#include "timeutils.h"
#include "trace.h"

#include <errno.h>
#include <string>
#include <vector>
#include <map>
#include <pthread.h>

/**
 * Log important Paxos states so that a node can recover from failures.
 */
class paxos_log {
public:
    void loginstance(unsigned instance, const std::string &val);
};

/**
 * Add object message of a particular instance into a bucket.
 */
template <class T>
class obj_bucket {
public:
    enum wait_status {
        OK,
        FORCED_QUIT,
        TIMEDOUT
    };

    obj_bucket();
    ~obj_bucket();

    void watch(unsigned instance, unsigned size);

    /**
     * Wait till the bucket has gathered the specified number of objects with
     * a pre-configured instance value or a timeout occurs. timeout is in
     * milliseconds.
     * \return if the bucket has got enough objects
     */
    int wait(int timeout=5000);

    /**
     * Force to quit waiting for the bucket to be full.
     */
    void stop_waiting();

    /**
     * Try to put an object into the bucket.
     */
    bool add(unsigned instance, const T& o);

    void clear();

    unsigned instance() const;

    int expecting_size() const;

    int current_size() const;

    std::vector<T>& items();

private:
    std::vector<T> bucket_;
    mutable pthread_mutex_t m_;
    pthread_cond_t size_cv_;
    unsigned instance_;
    unsigned size_;
    bool locked_;
};

template <class T>
obj_bucket<T>::obj_bucket()
    : locked_(true)
{
    pthread_mutex_init(&m_, NULL);
    pthread_cond_init(&size_cv_, NULL);
}

template <class T>
obj_bucket<T>::~obj_bucket()
{
    clear();
    pthread_mutex_destroy(&m_);
    pthread_cond_destroy(&size_cv_);
}

template <class T>
void obj_bucket<T>::watch(unsigned instance, unsigned size)
{
    pthread_mutex_lock(&m_);
    TDEBUG("watching bucket for #%u size: %u", instance, size);
    instance_ = instance;
    size_ = size;
    // clear bucket
    bucket_.clear();
    locked_ = false;
    pthread_mutex_unlock(&m_);
}

template <class T>
int obj_bucket<T>::wait(int timeout)
{
    int r = OK;
    pthread_mutex_lock(&m_);
    struct timespec now, ts;
    clock_gettime(CLOCK_REALTIME, &now);
    add_timespec(now, timeout, &ts);
    TDEBUG("waiting for bucket to be full (timeout: %d ms)", timeout);
    while (bucket_.size() < size_) {
        if (pthread_cond_timedwait(&size_cv_, &m_, &ts) == ETIMEDOUT) {
            r = TIMEDOUT;
            break;
        }
    }
    if (bucket_.size() >= size_) {
        r = OK;
    } else if (r == OK) {
        r = FORCED_QUIT;
    }

    // do not allow add() to be called from now on
    locked_ = true;
    pthread_mutex_unlock(&m_);
    return r;
}

template <class T>
void obj_bucket<T>::stop_waiting()
{
    pthread_mutex_lock(&m_);
    pthread_cond_signal(&size_cv_);
    pthread_mutex_unlock(&m_);
}

template <class T>
bool obj_bucket<T>::add(unsigned instance, const T& o)
{
    bool r = false;
    pthread_mutex_lock(&m_);
    if (!locked_ && instance == instance_) {
        bucket_.push_back(o);
        TDEBUG("now bucket size: %u/%u", bucket_.size(), size_);
        if (bucket_.size() >= size_) {
            TDEBUG("enough objects gathered in bucket: %u", size_);
            pthread_cond_signal(&size_cv_);
        }
        r = true;
    } else {
        TWARN("bucket cannot be updated right now");
    }
    pthread_mutex_unlock(&m_);
    return r;
}

template <class T>
void obj_bucket<T>::clear()
{
    pthread_mutex_lock(&m_);
    bucket_.clear();
    pthread_mutex_unlock(&m_);
}

template <class T>
unsigned obj_bucket<T>::instance() const
{
    return instance_;
}

template <class T>
int obj_bucket<T>::expecting_size() const
{
    return size_;
}

template <class T>
int obj_bucket<T>::current_size() const
{
    int s;
    pthread_mutex_lock(&m_);
    s = bucket_.size();
    pthread_mutex_unlock(&m_);
    return s;
}

template <class T>
std::vector<T> & obj_bucket<T>::items()
{
    return bucket_;
}

class paxos_callback {
public:
    paxos_callback() {}
    virtual ~paxos_callback() {}

    virtual void paxos_commit(unsigned instance, const std::string &chosen) = 0;
};

/**
 * This class implements a multi-instance paxos protocol. It assumes that a node
 * runs a proposer, an acceptor, and a learner at the same time.
 */
class paxos : public subscriber {
public:

    paxos(const std::string &primary, messaging *m, paxos_callback *cb);
    ~paxos();

    /**
     * Start a new Paxos instance.
     * \return the chosen value (not necessarily the same as the proposed one)
     */
    std::string start(const std::vector<std::string> &mems,
            const std::string &val);

private:
    std::string primary_;
    messaging *hermes_;

    typedef void (paxos::*ev_handler)(const DataMessage *);
    std::map<std::string, ev_handler> handlers_;

    unsigned instance_; // current instance
    std::vector<std::string> members_; // current participants
    Propnum my_prop_;
    std::vector<std::string> values_; // decided values
    bool behind_;

    Propnum pp_seen_; // highest prepare seen; used for handling prep req
    Propnum pa_me_; // highest proposal accepted by me; for handling prep req
    std::string va_me_; // my latest accepted value; for handling prep req

    obj_bucket<std::pair<std::string, PrepareResponse *> > pbox_;
    obj_bucket<AcceptResponse *> abox_; // container for received accept resp
    obj_bucket<std::string> dbox_;

    pthread_mutex_t m_;
    pthread_cond_t prepare_cv_; // if the prepare phase has finished
    pthread_cond_t accept_cv_; // if the accept phase has finished
    pthread_cond_t decide_cv_; // if the learn phase has finished
    pthread_cond_t behind_cv_; // if we are behind and need to catch up

    pthread_t catchup_th_;

    paxos_callback *cb_;

    paxos(const paxos &another) {}
    paxos& operator=(const paxos &) {return *this;}
    /**
     * Phase 1 of Paxos.
     * @param[in] the members involved in this paxos instance
     * @param[out] store the members that reply with ok
     * @param[out] the accepted value with highest proposal number
     */
    bool prepare(unsigned instance, const std::vector<std::string> &mems,
            std::vector<std::string> &accepts, std::string &va);

    /**
     * Phase 2 of Paxos.
     */
    bool accept(unsigned instance, const std::vector<std::string> &mems,
            const std::string &v);

    /**
     * Phase 3 of Paxos.
     */
    void decide(unsigned instance, const std::vector<std::string> &mems,
            const std::string &v);

    void commit(unsigned instance, const std::string &value);

    void catchup();

    /**
     * Handle phase 1 message of Paxos.
     */
    void handle_preparereq(const DataMessage *data);

    void handle_prepareres(const DataMessage *data);

    /**
     * Handle phase 2 message of Paxos.
     */
    void handle_acceptreq(const DataMessage *data);

    void handle_acceptres(const DataMessage *data);

    /**
     * Handle phase 3 message of Paxos.
     */
    void handle_decidereq(const DataMessage *data);

    void handle_decideres(const DataMessage *data);

    std::string subsubject(const std::string &subject);

    // inherited from message_listener
    void on_message(const DataMessage *data);
};

#endif
